package common.mq.sender.service.impl;

import java.util.HashMap;
import java.util.Map;

import org.apache.commons.httpclient.HttpMethod;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.springframework.beans.factory.annotation.Autowired;

import com.google.gson.Gson;

import common.mq.po.Message;
import common.mq.register.po.Consumer;
import common.mq.register.service.IRegisterService;
import common.mq.sender.http.HRequest;
import common.mq.sender.http.HResponse;
import common.mq.sender.service.ISenderService;
import common.mq.sender.thrift.TConsumerService;
import common.mq.sender.thrift.TRequest;
import common.mq.sender.thrift.TResponse;
import common.mq.util.HttpUtil;

public class SenderServiceImpl implements ISenderService {

	@Autowired
	IRegisterService registerService;

	public void setRegisterService(IRegisterService registerService) {
		this.registerService = registerService;
	}

	@Override
	public boolean send(int queueId, Message message) {
		// TODO Auto-generated method stub 出错打日志

		// consumerId 与 queueId为同一值
		Consumer consumer = registerService.getConsumer(queueId);

		int now = (int) System.currentTimeMillis() / 1000;
		synchronized (consumer) {
			if (now < consumer.getNextAcceptTime()) { // 当前时间小于消费者上次接收消失时设定下次可接收消息时间，拒绝接收消息
				return false;
			} else {
				consumer.setNextAcceptTime(now + consumer.getFrequency());
				;
			}
		}
		if (!checkConsumer(consumer)) {
			return false;
		}

		boolean result = false;
		int repeatCount = consumer.getRepeatCount();
		Map<String, String> content = consumer.getContent();
		do {
			if (content.get("protocol").equals("0")) { // http
				result = sendByHttp(consumer, message);
			} else {
				result = sendByThrift(consumer, message);
			}
			if (result) {
				return true;
			} else {
				repeatCount--;
			}
		} while (repeatCount >= 0);

		return false;
	}

	/**
	 * 检查消息必填项
	 * 
	 * @return
	 */
	private boolean checkConsumer(Consumer consumer) {
		Map<String, String> content = consumer.getContent();
		// 主机及协议（http,thrift）必须
		if (!content.containsKey("host") || !content.containsKey("protocol")) {
			return false;
		}
		// TODO 正则检查格式
		return true;
	}

	private boolean sendByHttp(Consumer consumer, Message message) {
		HttpMethod httpSender = consumer.getHttpSender();
		if (httpSender == null) {
			Map<String, String> content = consumer.getContent();
			boolean isGet = true;// 默认GET
			if (content.containsKey("method")) {// 0-get 1-post
				isGet = content.get("method").equals("0") ? true : false;
			}
			Map<String, String> params = new HashMap<String, String>();
			params.put("data", new Gson().toJson(new HRequest(message)));
			try {
				httpSender = HttpUtil.getMethod(
						content.get("host"),
						content.containsKey("port") ? Integer.parseInt(content
								.get("port")) : 80,
						content.containsKey("uri") ? content.get("uri") : "/",
						isGet, params);
			}  catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				return false;
			}
			consumer.setHttpSender(httpSender);
		}
		String response = null;
		try {
			response = new String(httpSender.getResponseBodyAsString()
					.getBytes("utf-8"), "utf-8");
			HResponse r = new Gson().fromJson(response, HResponse.class);
			// TODO 打印日志r.getMsg()
			return r.getCode() == 0 ? true : false;
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		}
	}

	private boolean sendByThrift(Consumer consumer, Message message) {
		TTransport thriftSender = consumer.getThriftSender();

		if (thriftSender == null) {
			Map<String, String> content = consumer.getContent();
			thriftSender = new TSocket(content.get("host"),
					content.containsKey("port") ? Integer.parseInt(content
							.get("port")) : 80);
			try {
				thriftSender.open();
			} catch (TTransportException e) {
				e.printStackTrace();
				return false;
			}
			thriftSender = new TFramedTransport(thriftSender);
			consumer.setThriftSender(thriftSender);
		}
		TProtocol protocol = new TBinaryProtocol(thriftSender);
		TConsumerService.Client client = new TConsumerService.Client(protocol);

		Map<String, String> content = message.getContent();
		TRequest request = new TRequest(content.get("caller"), message.getId(),
				message.getContent());
		TResponse r = null;
		try {
			r = client.receiveMessage(request);
		} catch (TException e) {
			e.printStackTrace();
			return false;
		}
		// TODO 打印日志r.getMsg()
		return r.getCode() == 0 ? true : false;
	}

}
